#include <unistd.h>     // pipe()
#include <pthread.h>
#include <event.h>
#include "k1_socket.h"
#include "k1_recv_thread.h"

struct client_conn {
    int fd;
    struct evbuffer *eb;
};


// ---------------- Global Var -------------------

// main thread var
//static thr_queue_t *workq;
static pthread_t tid;

// worker thread var
struct event_base *base;
int cmdpipe[2];
struct client_conn *connections;

// -------------- worker thread ---------------

static void
rt_worker_read_callback(int fd, short events, void *arg)
{
    char buf[80];
    ssize_t n;

    n = read(fd, buf, sizeof(buf)-1);
    printf("%d read %d ...\n", fd, n);
}

static void
rt_worker_listen_callback(int listenfd, short events, void *arg)
{
    int fd;
    struct event *ev = malloc(sizeof(struct event));

    fd = svr_accept(listenfd);
    printf("fd %d conn...\n", fd);

    event_set(ev, fd, EV_READ|EV_PERSIST, rt_worker_read_callback, ev);
    event_add(ev, NULL);
}

static void
rt_worker_init_listenfd()
{
    int listenfd;
    struct svr_opt opt = { K1S_INADDR_ANY, LISTEN_PORT, 5 };
    struct event *listen_ev;

    listenfd = svr_new_listen_tcp(&opt);
    printf("listenfd = %d\n", listenfd);

    listen_ev = malloc(sizeof(struct event));
    event_set(listen_ev, listenfd, EV_READ|EV_PERSIST, rt_worker_listen_callback, listen_ev);
    event_add(listen_ev, NULL);
}


// -------------- thread cmd pipe---------------

static void
rt_worker_cmdpipe_callback(int pipefd, short events, void *arg)
{
    char buf[80];
    ssize_t n;

    n = read(pipefd, buf, sizeof(buf)-1);
    buf[n] = '\0';

    printf("pipe cmd %d read = %s\n", pipefd, buf);
}

static void
rt_worker_init_cmdpipe()
{
    struct event *pipe_ev;

    pipe(cmdpipe);

    pipe_ev = malloc(sizeof(struct event));
    event_set(pipe_ev, cmdpipe[0], EV_READ|EV_PERSIST, rt_worker_cmdpipe_callback, pipe_ev);
    event_add(pipe_ev, NULL);
}


// -------------- thread init ---------------

static void
rt_worker_init()
{
    base = event_init();
    rt_worker_init_listenfd();
    rt_worker_init_cmdpipe();
}

static void *
rt_worker_main(void *_)
{
    rt_worker_init();
    event_dispatch();
    return NULL;
}


// -------------- init/shutdown ---------------

void recv_thread_init()
{
    // start thread
//    workq = thr_queue_create(max_online * max_online / 2);
    pthread_create(&tid, NULL, rt_worker_main, NULL);
    sleep(1);   // make pipe ok
}

void recv_thread_shutdown()
{
}

int recv_thread_cmd_new_conn_fd()
{
    write(cmdpipe[1], "hello", 5);
    return 0;
}


// -------------- recv thread cmd ---------------


